package com.daidai.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.daidai.util.{KafkaUtil, OffsetsUtil, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import redis.clients.jedis.Jedis

import java.util

object OdsBaseDbApp {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_db_app").setMaster("local[*]")
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    val topicName: String = "ODS_BASE_DB"
    val groupId: String = "ODS_BASE_DB_GROUP"

    // 从redis中读取偏移量
    val offsets: Map[TopicPartition, Long] = OffsetsUtil.readOffset(topicName, groupId)

    // 从Kafka中消费数据
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsets != null && offsets.nonEmpty) {
      kafkaDStream = KafkaUtil.getKafkaStream(topicName, ssc, offsets, groupId)
    } else {
      kafkaDStream = KafkaUtil.getKafkaStream(topicName, ssc, groupId)
    }

    // 提取偏移量结束点
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 处理数据
    val jsonObjDStream: DStream[JSONObject] = offsetRangesDStream.map(
      consumerRecord => {
        val dataJson: String = consumerRecord.value()
        val jSONObject: JSONObject = JSON.parseObject(dataJson)
        jSONObject
      }
    )

    jsonObjDStream.foreachRDD(
      rdd => {
        val redisFactKeys: String = "FACT:TABLES"
        val redisDimKeys: String = "DIM:TABLES"
        val jedis: Jedis = RedisUtil.getJedisFromPool()
        //事实表清单
        val factTables: util.Set[String] = jedis.smembers(redisFactKeys)
        println("factTables: " + factTables)
        //做成广播变量
        val factTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTables)
        //维度表清单
        val dimTables: util.Set[String] = jedis.smembers(redisDimKeys)
        println("dimTables: " + dimTables)
        //做成广播变量
        val dimTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTables)
        jedis.close()

        rdd.foreachPartition(
          jsonObjIter => {
            // 开启redis连接
            val jedis: Jedis = RedisUtil.getJedisFromPool()
            for (jsonObj <- jsonObjIter) {
              // 提取操作类型
              val operType: String = jsonObj.getString("type")
              val opValue: String = operType match {
                case "bootstrap-insert" => "I"
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null
              }
              //判断操作类型: 1. 明确什么操作  2. 过滤不感兴趣的数据
              if (opValue != null) {
                // 提取表名
                val tableName: String = jsonObj.getString("table")

                // 事实数据
                if (factTablesBC.value.contains(tableName)) {
                  val data: String = jsonObj.getString("data")
                  // DWD_ORDER_INFO_I  DWD_ORDER_INFO_U  DWD_ORDER_INFO_D
                  val dwdTopicName: String = s"DWD_${tableName.toUpperCase}_${opValue}"
                  KafkaUtil.send(dwdTopicName, data)

                  //模拟数据延迟
                  if (tableName.equals("order_detail")) {
                    Thread.sleep(200)
                  }
                }

                //维度数据
                if (dimTablesBC.value.contains(tableName)) {
                  val dataObj: JSONObject = jsonObj.getJSONObject("data")
                  val id: String = dataObj.getString("id")
                  val redisKey: String = s"DIM:${tableName.toUpperCase}:$id"
                  // 在此处开关redis的连接太频繁.
                  //val jedis: Jedis = MyRedisUtils.getJedisFromPool()
                  jedis.set(redisKey, dataObj.toJSONString)
                  //jedis.close()
                }
              }
            }
            //关闭redis连接
            jedis.close()
            //刷新Kafka缓冲区
            KafkaUtil.flush()
          }
        )
        //提交offset
        OffsetsUtil.saveOffset(topicName, groupId, offsetRanges)
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }
}
